package com.tz.test.service.impl;

import com.tz.test.entity.ScoreDTO;
import com.tz.test.service.InsertDataHouseService;
import com.tz.test.service.MultiThreadTransactionService;
import com.tz.test.thread.HandlerTask;
import com.tz.test.vo.RollBack;
import com.tz.test.vo.TransactionInfo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.*;

@Slf4j
@Service
public class MultiThreadTransactionServiceV1Impl implements MultiThreadTransactionService<ScoreDTO> {


    /**
     * 定义的线程数
     */
    private static final int THREAD_COUNT = Runtime.getRuntime().availableProcessors()*2+1 ;
    /**
     * 初始化线程池
     */
    private ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);


    @Resource
    private InsertDataHouseService insertDataHouseService;

    @Override
    public void handlerPartitionData(List<ScoreDTO> scoreList) {
        if (CollectionUtils.isEmpty(scoreList)) {
            log.warn("handleMessage data is null");
            return;
        }
        // 主线程监控
        CountDownLatch mainLatch = new CountDownLatch(1);
        // 子线程监控
        CountDownLatch threadLatch = null;
        // 事务共享管理器
        RollBack rollBack = new RollBack(false);
        BlockingDeque<Boolean> threadResult = new LinkedBlockingDeque<>(THREAD_COUNT);
        TransactionInfo transactionInfo = new TransactionInfo();
        transactionInfo.setThreadResult(threadResult);
        transactionInfo.setRollBack(rollBack);
        transactionInfo.setMainLatch(mainLatch);

        int batchSize = scoreList.size() / THREAD_COUNT;
        try {
            if (scoreList.size() < 20000) {
                insertDataHouseService.saveData(transactionInfo, scoreList);
            } else {
                // 拆分大小为batchSize一个存储
                transactionInfo.setMultiThreading(true);
                threadLatch = new CountDownLatch(THREAD_COUNT);
                transactionInfo.setThreadLatch(threadLatch);
                for (int i = 0; i < THREAD_COUNT; i++) {
                    List<ScoreDTO> partitionList;
                    if (i == THREAD_COUNT - 1){
                        partitionList = scoreList.subList(batchSize * i, scoreList.size());
                    }else {
                        partitionList = scoreList.subList(batchSize * i, (batchSize) * (i + 1));
                    }
                    HandlerTask<ScoreDTO> handlerTask = new HandlerTask<>(transactionInfo, partitionList, insertDataHouseService);
                    executor.submit(handlerTask);
                }
            }

            // 判断子线程是否成功
            if (threadLatch != null) {
                // 等待子线程执行完毕
                threadLatch.await();
                while (threadResult.size() > 0) {
                    //判断每条线程执行结果是否都正常
                    final Boolean re = threadResult.take();
                    if (!re) {
                        //子线程有异常
                        rollBack.setIsRollBack(true);
                        break;
                    }
                }
            }
        } catch (Exception e) {
            log.error("handlerPartitionData exception ", e);
            // 主线程发生异常 也要回滚
            rollBack.setIsRollBack(true);
        } finally {
            mainLatch.countDown();
            executor.shutdown();
        }
    }
}
